package online.shenjian.hadoop.mapjoin;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.yarn.webapp.hamlet2.Hamlet;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;

/**
 * @author shenjian
 * @since 2023/2/26
 */
public class OrderMapper extends Mapper<LongWritable, Text, Text, NullWritable> {

    private Map<String, String> productMap = new HashMap<>();
    private Text keyOut = new Text();

    @Override
    protected void setup(Mapper<LongWritable, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {
        // 获取缓存的文件小表product.txt
        URI[] cacheFiles = context.getCacheFiles();
        FSDataInputStream fsDataInputStream = FileSystem.get(context.getConfiguration()).open(new Path(cacheFiles[0]));
        BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(fsDataInputStream, "UTF-8"));
        String line;
        // 01 苹果
        while (StringUtils.isNotBlank(line = bufferedReader.readLine())) {
            String[] field = line.split(" ");
            productMap.put(field[0], field[1]);
        }
        IOUtils.closeStream(bufferedReader);
    }


    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String[] fields = line.split(" ");
        // 1001 01 1
        String productName = productMap.get(fields[1]);
        keyOut.set(fields[0] + "\t" + productName + "\t" + fields[2]);
        context.write(keyOut, NullWritable.get());
    }
}
